package sm.hibernate.persister;

import java.io.Serializable;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Iterator;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.hibernate.HibernateException;
import org.hibernate.MappingException;
import org.hibernate.TransientObjectException;
import org.hibernate.cache.CacheConcurrencyStrategy;
import org.hibernate.cache.CacheException;
import org.hibernate.cfg.Configuration;
import org.hibernate.collection.PersistentCollection;
import org.hibernate.engine.SessionFactoryImplementor;
import org.hibernate.engine.SessionImplementor;
import org.hibernate.exception.JDBCExceptionHelper;
import org.hibernate.jdbc.Expectation;
import org.hibernate.jdbc.Expectations;
import org.hibernate.loader.Loader;
import org.hibernate.loader.collection.BatchingCollectionInitializer;
import org.hibernate.loader.collection.CollectionInitializer;
import org.hibernate.mapping.Collection;
import org.hibernate.persister.collection.AbstractCollectionPersister;
import org.hibernate.persister.collection.BasicCollectionPersister;
import org.hibernate.pretty.MessageHelper;
import org.hibernate.type.ComponentType;
import org.hibernate.type.Type;
import org.hibernate.util.ArrayHelper;

import sm.hibernate.loader.SMCollectionLoader;
import sm.hibernate.sql.SMDelete;
import sm.hibernate.sql.SMSimpleSelect;
import sm.hibernate.sql.SMUpdate;

public class SMCollectionPersister extends BasicCollectionPersister {
	private static final Log log = LogFactory
			.getLog(AbstractCollectionPersister.class);

	private final String smsqlSelectSizeString;

	private final String smsqlSelectRowByIndexString;

	private final String smsqlDetectRowByIndexString;

	private final String smsqlDetectRowByElementString;

	private final int smBaseIndex = 1;

	public SMCollectionPersister(Collection collection,
			CacheConcurrencyStrategy cache, Configuration cfg,
			SessionFactoryImplementor factory) throws MappingException,
			CacheException {
		super(collection, cache, cfg, factory);
		smsqlSelectSizeString = generateSelectSizeString(collection.isIndexed()
				&& !collection.isMap());
		smsqlSelectRowByIndexString = generateSelectRowByIndexString();
		smsqlDetectRowByIndexString = generateDetectRowByIndexString();
		smsqlDetectRowByElementString = generateDetectRowByElementString();
	}

	@Override
	protected String generateDeleteString() {
		// customize: use SMDelete
		SMDelete delete = new SMDelete().setTableName(qualifiedTableName)
				.setPrimaryKeyColumnNames(keyColumnNames);

		if (hasWhere)
			delete.setWhere(sqlWhereString);

		if (getFactory().getSettings().isCommentsEnabled()) {
			delete.setComment("delete collection " + getRole());
		}

		return delete.toStatementString();
	}

	@Override
	protected CollectionInitializer createCollectionInitializer(
			java.util.Map enabledFilters) throws MappingException {
		if (batchSize > 1) {
			int[] batchSizesToCreate = ArrayHelper.getBatchSizes(batchSize);
			Loader[] loadersToCreate = new Loader[batchSizesToCreate.length];
			for (int i = 0; i < batchSizesToCreate.length; i++) {
				loadersToCreate[i] = new SMCollectionLoader(this,
						batchSizesToCreate[i], getFactory(), enabledFilters);
			}
			return new BatchingCollectionInitializer(this, batchSizesToCreate,
					loadersToCreate);
		} else {
			return new SMCollectionLoader(this, getFactory(), enabledFilters);
		}
	}

	// customize: handle null composite key, set ? twice
	private int writeNullableKey(Serializable id, SessionImplementor session,
			PreparedStatement st, int offset) throws SQLException {
		int loc = offset;
		Type identifierType = getKeyType();
		if (identifierType instanceof ComponentType) {
			ComponentType componentIdType = (ComponentType) identifierType;
			Type[] propertyTypes = componentIdType.getSubtypes();
			Object[] subvalues = componentIdType.getPropertyValues(id,
					session.getEntityMode());
			for (int i = 0; i < propertyTypes.length; i++) {
				int columnSpan = propertyTypes[i].getColumnSpan(session
						.getFactory());
				propertyTypes[i].nullSafeSet(st, subvalues[i], loc, session);
				loc += columnSpan;
				propertyTypes[i].nullSafeSet(st, subvalues[i], loc, session);
				loc += columnSpan;
			}
		} else {
			loc = writeKey(st, id, loc, session);
			loc = writeKey(st, id, loc, session);
		}
		return loc;
	}

	@Override
	public void remove(Serializable id, SessionImplementor session)
			throws HibernateException {

		if (!isInverse() && isRowDeleteEnabled()) {

			if (log.isDebugEnabled()) {
				log.debug("Deleting collection: "
						+ MessageHelper.collectionInfoString(this, id,
								getFactory()));
			}

			// Remove all the old entries

			try {
				int offset = 1;
				PreparedStatement st = null;
				Expectation expectation = Expectations
						.appropriateExpectation(getDeleteAllCheckStyle());
				boolean callable = isDeleteAllCallable();
				boolean useBatch = expectation.canBeBatched();
				String sql = getSQLDeleteString();
				if (useBatch) {
					if (callable) {
						st = session.getBatcher()
								.prepareBatchCallableStatement(sql);
					} else {
						st = session.getBatcher().prepareBatchStatement(sql);
					}
				} else {
					if (callable) {
						st = session.getBatcher().prepareCallableStatement(sql);
					} else {
						st = session.getBatcher().prepareStatement(sql);
					}
				}

				try {
					offset += expectation.prepare(st);

					// customize: handle null composite key, set ? twice
					offset = writeNullableKey(id, session, st, offset);

					if (useBatch) {
						session.getBatcher().addToBatch(expectation);
					} else {
						expectation.verifyOutcome(st.executeUpdate(), st, -1);
					}
				} catch (SQLException sqle) {
					if (useBatch) {
						session.getBatcher().abortBatch(sqle);
					}
					throw sqle;
				} finally {
					if (!useBatch) {
						session.getBatcher().closeStatement(st);
					}
				}

				if (log.isDebugEnabled()) {
					log.debug("done deleting collection");
				}
			} catch (SQLException sqle) {
				throw JDBCExceptionHelper.convert(
						getSQLExceptionConverter(),
						sqle,
						"could not delete collection: "
								+ MessageHelper.collectionInfoString(this, id,
										getFactory()), getSQLDeleteString());
			}
		}
	}

	@Override
	public void deleteRows(PersistentCollection collection, Serializable id,
			SessionImplementor session) throws HibernateException {

		if (!isInverse() && isRowDeleteEnabled()) {

			if (log.isDebugEnabled()) {
				log.debug("Deleting rows of collection: "
						+ MessageHelper.collectionInfoString(this, id,
								getFactory()));
			}

			boolean deleteByIndex = !isOneToMany() && hasIndex
					&& !indexContainsFormula;

			try {
				// delete all the deleted entries
				Iterator deletes = collection.getDeletes(this, !deleteByIndex);
				if (deletes.hasNext()) {
					int offset = 1;
					int count = 0;
					while (deletes.hasNext()) {
						PreparedStatement st = null;
						Expectation expectation = Expectations
								.appropriateExpectation(getDeleteCheckStyle());
						boolean callable = isDeleteCallable();
						boolean useBatch = expectation.canBeBatched();
						String sql = getSQLDeleteRowString();

						if (useBatch) {
							if (callable) {
								st = session.getBatcher()
										.prepareBatchCallableStatement(sql);
							} else {
								st = session.getBatcher()
										.prepareBatchStatement(sql);
							}
						} else {
							if (callable) {
								st = session.getBatcher()
										.prepareCallableStatement(sql);
							} else {
								st = session.getBatcher().prepareStatement(sql);
							}
						}

						try {
							expectation.prepare(st);

							Object entry = deletes.next();
							int loc = offset;
							if (hasIdentifier) {
								writeIdentifier(st, entry, loc, session);
							} else {
								// customize: handle null composite key, set ?
								// twice
								loc = writeNullableKey(id, session, st, loc);

								if (deleteByIndex) {
									writeIndexToWhere(st, entry, loc, session);
								} else {
									writeElementToWhere(st, entry, loc, session);
								}
							}

							if (useBatch) {
								session.getBatcher().addToBatch(expectation);
							} else {
								expectation.verifyOutcome(st.executeUpdate(),
										st, -1);
							}
							count++;
						} catch (SQLException sqle) {
							if (useBatch) {
								session.getBatcher().abortBatch(sqle);
							}
							throw sqle;
						} finally {
							if (!useBatch) {
								session.getBatcher().closeStatement(st);
							}
						}

						if (log.isDebugEnabled()) {
							log.debug("done deleting collection rows: " + count
									+ " deleted");
						}
					}
				} else {
					if (log.isDebugEnabled()) {
						log.debug("no rows to delete");
					}
				}
			} catch (SQLException sqle) {
				throw JDBCExceptionHelper.convert(
						getSQLExceptionConverter(),
						sqle,
						"could not delete collection rows: "
								+ MessageHelper.collectionInfoString(this, id,
										getFactory()), getSQLDeleteRowString());
			}
		}
	}

	@Override
	protected int doUpdateRows(Serializable id,
			PersistentCollection collection, SessionImplementor session)
			throws HibernateException {

		if (ArrayHelper.isAllFalse(elementColumnIsSettable))
			return 0;

		try {
			PreparedStatement st = null;
			Expectation expectation = Expectations
					.appropriateExpectation(getUpdateCheckStyle());
			boolean callable = isUpdateCallable();
			boolean useBatch = expectation.canBeBatched();
			Iterator entries = collection.entries(this);
			String sql = getSQLUpdateRowString();
			int i = 0;
			int count = 0;
			while (entries.hasNext()) {
				Object entry = entries.next();
				if (collection.needsUpdating(entry, i, elementType)) {
					int offset = 1;

					if (useBatch) {
						if (st == null) {
							if (callable) {
								st = session.getBatcher()
										.prepareBatchCallableStatement(sql);
							} else {
								st = session.getBatcher()
										.prepareBatchStatement(sql);
							}
						}
					} else {
						if (callable) {
							st = session.getBatcher().prepareCallableStatement(
									sql);
						} else {
							st = session.getBatcher().prepareStatement(sql);
						}
					}

					try {
						offset += expectation.prepare(st);
						int loc = writeElement(st,
								collection.getElement(entry), offset, session);
						if (hasIdentifier) {
							writeIdentifier(st,
									collection.getIdentifier(entry, i), loc,
									session);
						} else {
							// customize: handle null composite key, set ? twice
							loc = writeNullableKey(id, session, st, loc);

							if (hasIndex && !indexContainsFormula) {
								writeIndexToWhere(st,
										collection.getIndex(entry, i, this),
										loc, session);
							} else {
								writeElementToWhere(
										st,
										collection.getSnapshotElement(entry, i),
										loc, session);
							}
						}

						if (useBatch) {
							session.getBatcher().addToBatch(expectation);
						} else {
							expectation.verifyOutcome(st.executeUpdate(), st,
									-1);
						}
					} catch (SQLException sqle) {
						if (useBatch) {
							session.getBatcher().abortBatch(sqle);
						}
						throw sqle;
					} finally {
						if (!useBatch) {
							session.getBatcher().closeStatement(st);
						}
					}
					count++;
				}
				i++;
			}
			return count;
		} catch (SQLException sqle) {
			throw JDBCExceptionHelper.convert(
					getSQLExceptionConverter(),
					sqle,
					"could not update collection rows: "
							+ MessageHelper.collectionInfoString(this, id,
									getFactory()), getSQLUpdateRowString());
		}
	}

	@Override
	protected String generateUpdateRowString() {
		// customize: use SMUpdate
		SMUpdate update = new SMUpdate(getDialect())
				.setTableName(qualifiedTableName);

		// if ( !elementIsFormula ) {
		update.addColumns(elementColumnNames, elementColumnIsSettable);
		// }

		if (hasIdentifier) {
			update.setPrimaryKeyColumnNames(new String[] { identifierColumnName });
		} else if (hasIndex && !indexContainsFormula) {
			update.setPrimaryKeyColumnNames(keyColumnNames);
			update.setIndexColumnNames(indexColumnNames);
		} else {
			update.setPrimaryKeyColumnNames(ArrayHelper.join(keyColumnNames,
					elementColumnNames, elementColumnIsInPrimaryKey));
		}

		if (getFactory().getSettings().isCommentsEnabled()) {
			update.setComment("update collection row " + getRole());
		}

		return update.toStatementString();
	}

	@Override
	protected String generateDeleteRowString() {
		// customize: use SMDelete
		SMDelete delete = new SMDelete().setTableName(qualifiedTableName);

		if (hasIdentifier) {
			delete.setPrimaryKeyColumnNames(new String[] { identifierColumnName });
		} else if (hasIndex && !indexContainsFormula) {
			// TODO
			delete.setPrimaryKeyColumnNames(keyColumnNames);
			delete.setIndexColumnNames(indexColumnNames);
		} else {
			delete.setPrimaryKeyColumnNames(ArrayHelper.join(keyColumnNames,
					elementColumnNames, elementColumnIsInPrimaryKey));
		}

		if (getFactory().getSettings().isCommentsEnabled()) {
			delete.setComment("delete collection row " + getRole());
		}

		return delete.toStatementString();
	}

	@Override
	protected String generateSelectSizeString(boolean isIntegerIndexed) {
		String selectValue = isIntegerIndexed ? "max("
				+ getIndexColumnNames()[0] + ") + 1" : // lists, arrays
				"count(" + getElementColumnNames()[0] + ")"; // sets, maps, bags
		// customize: use SMSimpleSelect
		return new SMSimpleSelect(getDialect()).setTableName(getTableName())
				.addKey(getKeyColumnNames()).addColumn(selectValue)
				.toStatementString();
	}

	@Override
	protected String generateSelectRowByIndexString() {
		if (!hasIndex()) {
			return null;
		}

		// customize: use SMSimpleSelect
		return new SMSimpleSelect(getDialect()).setTableName(getTableName())
				.addKey(getKeyColumnNames())
				.addCondition(getIndexColumnNames(), "=?")
				.addCondition(indexFormulas, "=?")
				.addColumns(getElementColumnNames(), elementColumnAliases)
				.addColumns(indexFormulas, indexColumnAliases)
				.toStatementString();
	}

	@Override
	protected String generateDetectRowByIndexString() {
		if (!hasIndex()) {
			return null;
		}

		// customize: use SMSimpleSelect
		return new SMSimpleSelect(getDialect()).setTableName(getTableName())
				.addKey(getKeyColumnNames())
				.addCondition(getIndexColumnNames(), "=?")
				.addCondition(indexFormulas, "=?").addColumn("1")
				.toStatementString();
	}

	protected String generateDetectRowByElementString() {
		// customize: use SMSimpleSelect
		return new SMSimpleSelect(getDialect()).setTableName(getTableName())
				.addKey(getKeyColumnNames())
				.addCondition(getElementColumnNames(), "=?")
				.addCondition(elementFormulas, "=?").addColumn("1")
				.toStatementString();
	}

	@Override
	public int getSize(Serializable key, SessionImplementor session) {
		try {
			PreparedStatement st = session.getBatcher().prepareSelectStatement(
					smsqlSelectSizeString);
			try {

				writeNullableKey(key, session, st, 1);
				ResultSet rs = st.executeQuery();
				try {
					return rs.next() ? rs.getInt(1) - smBaseIndex : 0;
				} finally {
					rs.close();
				}
			} finally {
				session.getBatcher().closeStatement(st);
			}
		} catch (SQLException sqle) {
			throw JDBCExceptionHelper.convert(
					getFactory().getSQLExceptionConverter(),
					sqle,
					"could not retrieve collection size: "
							+ MessageHelper.collectionInfoString(this, key,
									getFactory()), smsqlSelectSizeString);
		}
	}

	public Object getElementByIndex(Serializable key, Object index,
			SessionImplementor session, Object owner) {
		try {
			PreparedStatement st = session.getBatcher().prepareSelectStatement(
					smsqlSelectRowByIndexString);
			try {
				int offset = writeNullableKey(key, session, st, 1);
				getIndexType().nullSafeSet(st, incrementIndexByBase(index),
						offset, session);
				ResultSet rs = st.executeQuery();
				try {
					if (rs.next()) {
						return getElementType().nullSafeGet(rs,
								elementColumnAliases, session, owner);
					} else {
						return null;
					}
				} finally {
					rs.close();
				}
			} finally {
				session.getBatcher().closeStatement(st);
			}
		} catch (SQLException sqle) {
			throw JDBCExceptionHelper.convert(
					getFactory().getSQLExceptionConverter(),
					sqle,
					"could not read row: "
							+ MessageHelper.collectionInfoString(this, key,
									getFactory()), smsqlSelectSizeString);
		}
	}

	public boolean indexExists(Serializable key, Object index,
			SessionImplementor session) {
		try {
			PreparedStatement st = session.getBatcher().prepareSelectStatement(
					smsqlDetectRowByIndexString);
			try {
				int offset = writeNullableKey(key, session, st, 1);
				getIndexType().nullSafeSet(st, incrementIndexByBase(index),
						offset, session);
				ResultSet rs = st.executeQuery();
				try {
					return rs.next();
				} finally {
					rs.close();
				}
			} catch (TransientObjectException e) {
				return false;
			} finally {
				session.getBatcher().closeStatement(st);
			}
		} catch (SQLException sqle) {
			throw JDBCExceptionHelper.convert(
					getFactory().getSQLExceptionConverter(),
					sqle,
					"could not check row existence: "
							+ MessageHelper.collectionInfoString(this, key,
									getFactory()), smsqlSelectSizeString);
		}
	}

	public boolean elementExists(Serializable key, Object element,
			SessionImplementor session) {
		try {
			PreparedStatement st = session.getBatcher().prepareSelectStatement(
					smsqlDetectRowByElementString);
			try {
				int offset = writeNullableKey(key, session, st, 1);
				getElementType().nullSafeSet(st, element, offset, session);
				ResultSet rs = st.executeQuery();
				try {
					return rs.next();
				} finally {
					rs.close();
				}
			} catch (TransientObjectException e) {
				return false;
			} finally {
				session.getBatcher().closeStatement(st);
			}
		} catch (SQLException sqle) {
			throw JDBCExceptionHelper.convert(
					getFactory().getSQLExceptionConverter(),
					sqle,
					"could not check row existence: "
							+ MessageHelper.collectionInfoString(this, key,
									getFactory()), smsqlSelectSizeString);
		}
	}

}
